package com.gongfutrip.trans.remoting;

import com.gongfutrip.trans.remoting.common.Pair;
import com.gongfutrip.trans.remoting.common.RemotingHelper;
import com.gongfutrip.trans.remoting.expception.RemotingSendRequestException;
import com.gongfutrip.trans.remoting.expception.RemotingTimeoutException;
import com.gongfutrip.trans.remoting.expception.RemotingTooMuchRequestException;
import com.gongfutrip.trans.remoting.netty.NettyEvent;
import com.gongfutrip.trans.remoting.netty.NettyEventType;
import com.gongfutrip.trans.remoting.netty.NettyServerConfig;
import com.gongfutrip.trans.remoting.protocol.RemotingCommand;
import com.gongfutrip.trans.util.RemotingUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import java.net.InetSocketAddress;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class NettyRemotingServer extends NettyRemotingAbstract implements RemotingServer {

    private static final Logger log = LoggerFactory.getLogger(NettyRemotingServer.class);
    private final ServerBootstrap serverBootstrap;
    private final NettyServerConfig nettyServerConfig;
    private final ChannelEventListener channelEventListener;
    private final ExecutorService publicExecutor;
    private final EventLoopGroup eventLoopGroupBoss;
    private final EventLoopGroup eventLoopGroupSelector;
    private DefaultEventExecutorGroup defaultEventExecutorGroup;
    private RPCHook rpcHook;
    private int port = 0;
    private final Timer timer = new Timer("ServerHouseKeepingService", true);

    public NettyRemotingServer(final NettyServerConfig nettyServerConfig) {
        this(nettyServerConfig, null);
    }

    public NettyRemotingServer(final NettyServerConfig nettyServerConfig,
        final ChannelEventListener channelEventListener) {
        // 设置单向调用和异步调用两种模式下的最大并发数 65535个
        super(nettyServerConfig.getServerOnewaySemaphoreValue(),
            nettyServerConfig.getServerAsyncSemaphoreValue());
        this.serverBootstrap = new ServerBootstrap();
        this.nettyServerConfig = nettyServerConfig;
        this.channelEventListener = channelEventListener;
        // 执行responseFuture回调函数的线程数 默认cpu数量
        int publicThreadNums = getPublicThreadNums(nettyServerConfig);
        // 执行responseFuture回调函数的线程池
        this.publicExecutor = getPublicExecutor(publicThreadNums);
        // 配置服务端的NIO线程组
        this.eventLoopGroupBoss = getEventLoopGroupBoss();

        this.eventLoopGroupSelector = getEventLoopGroupWorker(nettyServerConfig);

        // I／O线程是不允许被阻塞的，也就是不能在ChannelHandler中进行任何阻塞式的处理，但是对此我们也有相应的解决方法.
        // 就是在把ChannelHanders添加到ChannelPipeline的时候，
        // 指定一个EventExecutorGroup，ChannelHandler中所有的方法都将会在这个指定的EventExecutorGroup中运行。
        this.defaultEventExecutorGroup = buildDefaultEventExecutorGroup();

    }

    private EventLoopGroup getEventLoopGroupWorker(final NettyServerConfig nettyServerConfig) {
        // 如果是Linux平台，并且开启了native epoll，就用EpollEventLoopGroup，这个也就是用JNI，调的c写的epoll,否则，就用Java
        // NIO的NioEventLoopGroup。
        EventLoopGroup eventLoopGroupSelector = null;
        if (RemotingUtil.isLinuxPlatform() && nettyServerConfig.isUseEpollNativeSelector()) {
            eventLoopGroupSelector = new EpollEventLoopGroup(
                nettyServerConfig.getServerSelectorThreads(), new ThreadFactory() {
                private AtomicInteger threadIndex = new AtomicInteger(0);
                private int threadTotal = nettyServerConfig.getServerSelectorThreads();

                public Thread newThread(Runnable r) {
                    return new Thread(r,
                        String.format("NettyServerEPOLLSelector_%d_%d", threadTotal,
                            this.threadIndex.incrementAndGet()));
                }
            });
        } else {

            eventLoopGroupSelector =
                new NioEventLoopGroup(nettyServerConfig.getServerSelectorThreads(),
                    new ThreadFactory() {
                        private AtomicInteger threadIndex = new AtomicInteger(0);
                        private int threadTotal = nettyServerConfig.getServerSelectorThreads();

                        public Thread newThread(Runnable r) {
                            return new Thread(r,
                                String.format("NettyServerNIOSelector_%d_%d", threadTotal,
                                    this.threadIndex.incrementAndGet()));
                        }
                    });

        }
        return eventLoopGroupSelector;
    }

    /**
     * 注册rpc钩子
     */
    @Override
    public void registerRPCHook(RPCHook rpcHook) {
        this.rpcHook = rpcHook;
    }

    @Override
    public RPCHook getRPCHook() {
        return this.rpcHook;
    }

    private NioEventLoopGroup getEventLoopGroupBoss() {
        return new NioEventLoopGroup(1, new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "NettyBoss_" + threadIndex.incrementAndGet());
            }
        });
    }

    private ExecutorService getPublicExecutor(int publicThreadNums) {
        return Executors.newFixedThreadPool(publicThreadNums, new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r,
                    "NettyServerPublicExecutor_" + this.threadIndex.incrementAndGet());
            }
        });
    }

    private int getPublicThreadNums(final NettyServerConfig nettyServerConfig) {
        int publicThreadNums = nettyServerConfig.getServerCallbackExecutorThreads();
        if (publicThreadNums <= 0) {
            publicThreadNums = 4;
        }
        return publicThreadNums;
    }

    @Override
    public void start() {
        //http://www.cnblogs.com/ggjucheng/archive/2012/02/17/2355874.html
        ServerBootstrap childHandler =
            this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 1024)//客户端连接请求等待队列
                .option(ChannelOption.SO_REUSEADDR, true)//允许重复使用本地地址和端口
                .option(ChannelOption.SO_KEEPALIVE, false)//是否启用心跳保活机制
                .childOption(ChannelOption.TCP_NODELAY, true)//Nagle算法,将小的数据包组装为更大的帧然后进行发送
                .option(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())
                .option(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())
                .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))// 设置端口
                .childHandler(new ChannelInitializer<SocketChannel>() {

                    public void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(
                            defaultEventExecutorGroup,
                            new NettyEncoder(),
                            new NettyDecoder(),
                            new IdleStateHandler(0, 0,
                                nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                            new NettyConnetManageHandler(),
                            new NettyServerHandler());
                    }
                });
        // 是否使用Netty内存池
        if (nettyServerConfig.isServerPooledByteBufAllocatorEnable()) {
            childHandler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        }
        try {
            ChannelFuture sync = this.serverBootstrap.bind().sync();
            InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
            this.port = addr.getPort();
        } catch (InterruptedException e1) {
            throw new RuntimeException("this.serverBootstrap.bind().sync() InterruptedException",
                e1);
        }
        // 监听new NettyConnetManageHandler()产生的NettyEvent
        if (this.channelEventListener != null) {
            this.nettyEventExecuter.start();
        }
        // 定时扫描responseTable,获取返回结果,并且处理超时

        this.timer.scheduleAtFixedRate(new TimerTask() {

            public void run() {
                try {
                    NettyRemotingServer.this.scanResponseTable();
                } catch (Exception e) {
                    log.error("scanResponseTable exception", e);
                }
            }
        }, 1000 * 3, 1000);

    }

    /**
     * 用于处理netty handler中的操作 编码解码等
     */
    private DefaultEventExecutorGroup buildDefaultEventExecutorGroup() {
        return new DefaultEventExecutorGroup(nettyServerConfig.getServerWorkerThreads(),
            new ThreadFactory() {

                private AtomicInteger threadIndex = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r,
                        "NettyServerCodecThread_" + this.threadIndex.incrementAndGet());
                }
            });
    }

    @Override
    public void shutdown() {
        try {
            if (this.timer != null) {
                this.timer.cancel();
            }

            this.eventLoopGroupBoss.shutdownGracefully();

            this.eventLoopGroupSelector.shutdownGracefully();

            if (this.nettyEventExecuter != null) {
                this.nettyEventExecuter.shutdown();
            }

            if (this.defaultEventExecutorGroup != null) {
                this.defaultEventExecutorGroup.shutdownGracefully();
            }
        } catch (Exception e) {
            log.error("NettyRemotingServer shutdown exception, ", e);
        }

        if (this.publicExecutor != null) {
            try {
                this.publicExecutor.shutdown();
            } catch (Exception e) {
                log.error("NettyRemotingServer shutdown exception, ", e);
            }
        }
    }

    class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {

        protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg)
            throws Exception {
            processMessageReceived(ctx, msg);
        }

    }

    /**
     * Netty框架内的全部通信相关的事件
     *
     * @author liuchengyu
     */
    class NettyConnetManageHandler extends ChannelDuplexHandler {

        @Override /* 记录日志 */
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.info("NETTY SERVER PIPELINE: channelRegistered {}", remoteAddress);
            super.channelRegistered(ctx);
        }

        @Override /* 记录日志 */
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.info("NETTY SERVER PIPELINE: channelUnregistered, the channel[{}]", remoteAddress);
            super.channelUnregistered(ctx);
        }

        @Override /* 生成*NettyEvent 放入LinkedBlockingQueue队列 */
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.info("NETTY SERVER PIPELINE: channelActive, the channel[{}]", remoteAddress);
            super.channelActive(ctx);

            if (NettyRemotingServer.this.channelEventListener != null) {
                NettyRemotingServer.this.putNettyEvent(
                    new NettyEvent(NettyEventType.CONNECT, remoteAddress.toString(),
                        ctx.channel()));
            }
        }

        @Override /* 生成*NettyEvent 放入LinkedBlockingQueue队列 */
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.info("NETTY SERVER PIPELINE: channelInactive, the channel[{}]", remoteAddress);
            super.channelInactive(ctx);

            if (NettyRemotingServer.this.channelEventListener != null) {
                NettyRemotingServer.this.putNettyEvent(
                    new NettyEvent(NettyEventType.CLOSE, remoteAddress.toString(), ctx.channel()));
            }
        }

        @Override /* 生成*NettyEvent 放入LinkedBlockingQueue队列 */
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent evnet = (IdleStateEvent) evt;
                if (evnet.state().equals(IdleState.ALL_IDLE)) {
                    final String remoteAddress = RemotingHelper
                        .parseChannelRemoteAddr(ctx.channel());
                    log.warn("NETTY SERVER PIPELINE: IDLE exception [{}]", remoteAddress);
                    RemotingUtil.closeChannel(ctx.channel());
                    if (NettyRemotingServer.this.channelEventListener != null) {
                        NettyRemotingServer.this.putNettyEvent(
                            new NettyEvent(NettyEventType.IDLE, remoteAddress.toString(),
                                ctx.channel()));
                    }
                }
            }

            ctx.fireUserEventTriggered(evt);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.warn("NETTY SERVER PIPELINE: exceptionCaught {}", remoteAddress);
            log.warn("NETTY SERVER PIPELINE: exceptionCaught exception.", cause);

            if (NettyRemotingServer.this.channelEventListener != null) {
                NettyRemotingServer.this.putNettyEvent(
                    new NettyEvent(NettyEventType.EXCEPTION, remoteAddress.toString(),
                        ctx.channel()));
            }

            RemotingUtil.closeChannel(ctx.channel());
        }
    }

    @Override
    public ChannelEventListener getChannelEventListener() {
        return channelEventListener;
    }

    /**
     * 注册NettyRequestProcessor用来处理接收请求
     */
    @Override
    public void registerProcessor(int requestCode, NettyRequestProcessor processor,
        ExecutorService executor) {
        ExecutorService executorThis = executor;
        if (null == executor) {
            executorThis = this.publicExecutor;
        }
        Pair<NettyRequestProcessor, ExecutorService> pair =
            new Pair<>(processor, executorThis);
        this.processorTable.put(requestCode, pair);
    }

    @Override
    public void registerDefaultProcessor(NettyRequestProcessor processor,
        ExecutorService executor) {
        this.defaultRequestProcessor =
            new Pair<NettyRequestProcessor, ExecutorService>(processor, executor);

    }

    @Override
    public int localListenPort() {
        return this.port;
    }

    @Override
    public Pair<NettyRequestProcessor, ExecutorService> getProcessorPair(int requestCode) {
        return processorTable.get(requestCode);
    }

    @Override
    public RemotingCommand invokeSync(Channel channel, RemotingCommand request, long timeoutMillis)
        throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException {
        return this.invokeSyncImpl(channel, request, timeoutMillis);
    }

    @Override
    public void invokeAsync(Channel channel, RemotingCommand request, long timeoutMillis,
        InvokeCallback invokeCallback) throws InterruptedException, RemotingTooMuchRequestException,
        RemotingTimeoutException, RemotingSendRequestException {
        this.invokeAsyncImpl(channel, request, timeoutMillis, invokeCallback);

    }

    @Override
    public void invokeOneway(Channel channel, RemotingCommand request, long timeoutMillis)
        throws InterruptedException, RemotingTooMuchRequestException, RemotingTimeoutException,
        RemotingSendRequestException {
        this.invokeOnewayImpl(channel, request, timeoutMillis);

    }

    @Override
    public ExecutorService getCallbackExecutor() {
        return this.publicExecutor;
    }

}


